package org.future.task.redis.cluster;

import org.future.task.core.serialize.MineSerializer;
import org.future.task.core.util.ThreadPoolUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Pipeline;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class CommandManager implements CommandClient<CEntry> {
    //创建JedisCluster
    //计算slot
    //存储并管理slot中的本地缓存及队列
    //批量提交

    Logger log = LoggerFactory.getLogger(this.getClass());

    protected JedisCluster jedisCluster;

    protected final ConcurrentHashMap<String, LinkedBlockingQueue<CEntry>> clientMapping = new ConcurrentHashMap<>();

    protected ConcurrentHashMap<String, String> slotToClient = new ConcurrentHashMap<>();

    private List<CommandThread> commandThreads = new ArrayList<>();

    protected ThreadPoolExecutor executorService;

    protected Command command;

    protected MineSerializer mineSerializer;

    protected int commandThreadSize;

    protected int localQueueSize;

    protected int batchSize;

    public int getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    public int getLocalQueueSize() {
        return localQueueSize;
    }

    public void setLocalQueueSize(int localQueueSize) {
        this.localQueueSize = localQueueSize;
    }

    public int getCommandThreadSize() {
        return commandThreadSize;
    }

    public void setCommandThreadSize(int commandThreadSize) {
        this.commandThreadSize = commandThreadSize;
    }

    public MineSerializer getMineSerializer() {
        return mineSerializer;
    }

    public void setMineSerializer(MineSerializer mineSerializer) {
        this.mineSerializer = mineSerializer;
    }

    public CommandManager(ClusterFactory clusterFactory, Command command, int commandThreadSize, int localQueueSize) {
        this.commandThreadSize = commandThreadSize;
        if (command == null) {
            command = new DefaultCommandImpl();
        }
        this.command = command;
        this.localQueueSize = localQueueSize;

        this.jedisCluster = clusterFactory.builder();
        restartCommitThread(null);
    }

    public CommandManager(ClusterFactory clusterFactory, int commandThreadSize, int localQueueSize) {
        this(clusterFactory, null, commandThreadSize, localQueueSize);
    }

    public CommandManager(ClusterFactory clusterFactory, int commandThreadSize) {
        this(clusterFactory, null, commandThreadSize, 0);
    }


    private void restartCommitThread(String slotSingle) {
        if (commandThreadSize <= 0) {
            commandThreadSize = 1;
        }
        Map<String, JedisPool> clusterNodes = jedisCluster.getClusterNodes();

        int size = jedisCluster.getClusterNodes().size();

        if (executorService == null) {
            synchronized (this) {
                if (executorService == null)
                    executorService = new ThreadPoolUtils().getDefaultThreadPool("cluster", commandThreadSize * size);
            }
        } else {
            synchronized (executorService) {
                executorService.setCorePoolSize(commandThreadSize * size);
            }
        }

        if (!commandThreads.isEmpty()) {
            for (CommandThread commandThread : commandThreads) {
                commandThread.setStop(true);
            }
            commandThreads.clear();
        }


        for (Map.Entry<String, JedisPool> stringJedisPoolEntry : clusterNodes.entrySet()) {
            String key = stringJedisPoolEntry.getKey();
            JedisPool value = stringJedisPoolEntry.getValue();

            if (slotSingle != null && slotSingle.equalsIgnoreCase(key)) {

                clientMapping.computeIfAbsent(slotSingle, (name) -> {

                    if (localQueueSize > 0)
                        return new LinkedBlockingQueue<>(localQueueSize);
                    else
                        return new LinkedBlockingQueue<>();
                });
            }

            for (int i = 0; i < commandThreadSize; i++) {
                CommandThread commandThread = new CommandThread(key, value);
                commandThreads.add(commandThread);
                executorService.execute(commandThread);
            }
        }

    }

    class CommandThread implements Runnable {

        private boolean stop = false;

        private String slotSingle;

        private JedisPool jedisPool;

        public CommandThread(String slotSingle, JedisPool jedisPool) {
            this.slotSingle = slotSingle;
            this.jedisPool = jedisPool;
        }

        public boolean isStop() {
            return stop;
        }

        public void setStop(boolean stop) {
            this.stop = stop;
        }

        @Override
        public void run() {

            boolean batchStatus = false;
            while (!stop) {

                Jedis resource = null;
                Pipeline pipelined = null;


                LinkedBlockingQueue<CEntry> cEntries = clientMapping.computeIfAbsent(slotSingle, (name) -> {

                    if (localQueueSize > 0)
                        return new LinkedBlockingQueue<>(localQueueSize);
                    else
                        return new LinkedBlockingQueue<>();
                });

                List<CEntry> cEntryList = new ArrayList<>();
                try {

                    if (!batchStatus) {
                        CEntry take = cEntries.take();
                        batchStatus = true;
                        cEntryList.add(take);
                    } else {
                        cEntries.drainTo(cEntryList, batchSize == 0 ? 200 : batchSize);
                    }

                    if (cEntryList.isEmpty()) {
                        batchStatus = false;
                        continue;
                    }
                    resource = jedisPool.getResource();
                    pipelined = resource.pipelined();
                    for (CEntry cEntry : cEntryList) {
                        try {
                            command.exec(pipelined, cEntry);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    log.info("插槽{}数据写入量:{}", slotSingle, cEntryList.size());

                } catch (Exception e) {
                    //TODO 出现异常  集群出现扩容或缩容 资源释放问题  缩容后 数据如何绑定
                    //TODO  方案：出现扩缩容后  摘除数据  并将摘除的数据集重新提交给一个新的写入线程 写入完毕后结束线程
                    //先加锁摘除  删除映射
                    LinkedBlockingQueue<CEntry> cEntryLbk;
                    synchronized (clientMapping) {
                        cEntryLbk = clientMapping.remove(slotSingle);
                        slotToClient.clear();
                        restartCommitThread(slotSingle);
                    }
                    Executors.newSingleThreadExecutor().submit(new FlushData(cEntryLbk));
                    for (CEntry cEntry : cEntryList) {
                        try {
                            putToRedis(cEntry);
                        } catch (InterruptedException e1) {
                            e1.printStackTrace();
                        }
                    }
                } finally {
                    if (pipelined != null) {
                        pipelined.close();
                    }
                    if (resource != null) {
                        resource.close();
                    }
                }

            }
            log.info("线程:{}结束设置为false", Thread.currentThread().getName());
        }
    }

    class FlushData implements Runnable {

        private LinkedBlockingQueue<CEntry> cEntries;

        public FlushData(LinkedBlockingQueue<CEntry> cEntries) {
            this.cEntries = cEntries;
        }

        @Override
        public void run() {
            while (cEntries.size() != 0) {
                for (CEntry cEntry : cEntries) {
                    try {
                        putToRedis(cEntry);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }

        }
    }
}
